/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.io.FileInputStream;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Properties;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

public class CSBServer {
	private static HashMap<String, Long> tps;
	private static boolean do_perf = false;
	
    private static Logger LOG = Logger.getLogger(CSBServer.class.getName());

    public static void main(String[] args) throws Exception {    	
    	PropertyConfigurator.configure("conf/log4j.properties");
    	
    	String fn = "conf/csbserver.properties";
    	if (args.length > 0)
    		fn = args[0];
    	Properties prop = new Properties();
    	FileInputStream fis = new FileInputStream(fn);
    	prop.load(fis);
    	fis.close();

    	String type = prop.getProperty("server.type", "");
    	String zkConn = prop.getProperty("zk.connect", "localhost:2181");
    	String srvConn = prop.getProperty("server.connect", "tcp://localhost:7001");
    	String kfkConn = prop.getProperty("kfk.connect", "inproc://InProc");
    	int numThread = Integer.parseInt(prop.getProperty("thread.number", "1"));
    	int perfInterval = Integer.parseInt(prop.getProperty("perf.interval", "60000"));
    	String perfClass = prop.getProperty("perf.monitor", "com.qloudgen.csb.CSBPerfMonitor");

    	CSBPerfMonitor monitor = null;
    	if (args.length > 1 && args[1].equals("perf")) {
    		do_perf = true;
        	monitor = (CSBPerfMonitor)
        		Class.forName(perfClass).newInstance();
    	}
    	   
    	if (type.equals("sender")) {
    		LOG.info("starting sender server...");
        	int ackMode = Integer.parseInt(prop.getProperty("ack.mode", "-1"));
    		CSBSender csbSender = new CSBSender(zkConn, srvConn, 
    				kfkConn, numThread, ackMode, perfInterval);
    		if (do_perf)
    			perf(csbSender, perfInterval, monitor);
    		csbSender.run();
    	}  else if (type.equals("queue")) {
    		LOG.info("starting queue server...");
    		String topic = prop.getProperty("topic", "csb");
    		String groupId = prop.getProperty("group.id", "csbgroup");
    		int bufferSize = Integer.parseInt(prop.getProperty("buffer.size", "10"));
    		int timeout = Integer.parseInt(prop.getProperty("consumer.timeout", "3000"));
    		int retry = Integer.parseInt(prop.getProperty("consumer.retry", "3"));
    		int mode = Integer.parseInt(prop.getProperty("fetch.mode", "0"));
        	String errClass = prop.getProperty("error.handler", "com.qloudgen.csb.CSBErrorHandler");
        	
        	CSBErrorHandler errHandler = (CSBErrorHandler)
        			Class.forName(errClass).newInstance();
        	
        	CSBQueue csbQueue = new CSBQueue(topic, groupId, 
    			zkConn, kfkConn, srvConn, numThread, bufferSize, 
    				timeout, retry, mode, errHandler, perfInterval);
    		if (do_perf)
    			perf(csbQueue, perfInterval, monitor);
    		csbQueue.run();
    	}
    	
    	System.exit(0);
    }

    private static void perf(final CSBServerBase server, 
    		final int interval, final CSBPerfMonitor monitor) {  
		new Thread(new Runnable() {
			@Override
			public void run() {	
				while (true) {
					try {
						Thread.sleep(interval);
					} catch (InterruptedException e) {}
					
					tps = server.getTPS();
					Iterator<String> itr = tps.keySet().iterator();
					long total = 0;
					while (itr.hasNext()) {
						String s = itr.next();
						total = total + tps.get(s);
						monitor.perf_mark(tps.get(s), s);
					}
					monitor.perf_mark(total, null);
				}
			}
		}).start();
    }
}

